package org.budo.dubbo.protocol.async.repository.kafka;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.BytesDeserializer;
import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.budo.dubbo.protocol.async.repository.AbstractAsyncRepository;
import org.budo.support.lang.util.MapUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
public abstract class AbstractKafkaAsyncRepository extends AbstractAsyncRepository {
    private static final String GROUP_ID = "group.id";

    private static final String ENABLE_AUTO_COMMIT = "enable.auto.commit";

    protected static final int POLL_TIMEOUT = 100;

    private static final Map<String, KafkaProducer<String, Bytes>> _KAFKA_PRODUCER_MAP = new ConcurrentHashMap<String, KafkaProducer<String, Bytes>>();

    private static final Map<String, KafkaConsumer<String, Bytes>> _KAFKA_CONSUMER_MAP = new ConcurrentHashMap<String, KafkaConsumer<String, Bytes>>();

    private Map<String, Object> properties = new HashMap<String, Object>();

    public AbstractKafkaAsyncRepository() {
        this.initDefaultProperties();
    }

    /**
     * 默认配置
     */
    private void initDefaultProperties() {
        properties.put(ENABLE_AUTO_COMMIT, "false"); // 不自动提交
        properties.put(GROUP_ID, this.getClass().getName());

        properties.put("key.deserializer", StringDeserializer.class.getName());
        properties.put("value.deserializer", BytesDeserializer.class.getName());

        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", BytesSerializer.class.getName());
    }

    public KafkaConsumer<String, Bytes> getKafkaConsumer(String destinationName) {
        KafkaConsumer<String, Bytes> kafkaConsumer = _KAFKA_CONSUMER_MAP.get(destinationName);
        if (null != kafkaConsumer) {
            return kafkaConsumer;
        }

        synchronized (this) {
            kafkaConsumer = _KAFKA_CONSUMER_MAP.get(destinationName);
            if (null != kafkaConsumer) {
                return kafkaConsumer;
            }

            // init KafkaConsumer
            Map<String, Object> properties = this.getProperties();
            Map<String, Object> consumerProperties = MapUtil.append(properties, //
                    GROUP_ID, properties.get(GROUP_ID) + "-" + destinationName); // 每个接口一个Consumer，一个GROUP_ID

            KafkaConsumer<String, Bytes> initKafkaConsumer = new KafkaConsumer<String, Bytes>(consumerProperties);

            initKafkaConsumer.subscribe(Arrays.asList(destinationName));

            _KAFKA_CONSUMER_MAP.put(destinationName, initKafkaConsumer);

            log.info("#59 initKafkaConsumer, destinationName=" + destinationName + ", initKafkaConsumer=" + initKafkaConsumer + ", properties=" + properties);
            return initKafkaConsumer;
        }
    }

    public KafkaProducer<String, Bytes> getKafkaProducer(String destinationName) {
        KafkaProducer<String, Bytes> kafkaProducer = _KAFKA_PRODUCER_MAP.get(destinationName);
        if (null != kafkaProducer) {
            return kafkaProducer;
        }

        synchronized (this) {
            kafkaProducer = _KAFKA_PRODUCER_MAP.get(destinationName);
            if (null != kafkaProducer) {
                return kafkaProducer;
            }

            // init KafkaProducer
            Map<String, Object> properties = this.getProperties();
            KafkaProducer<String, Bytes> initKafkaProducer = new KafkaProducer<String, Bytes>(properties);

            _KAFKA_PRODUCER_MAP.put(destinationName, initKafkaProducer);

            log.info("#102 initKafkaProducer, destinationName=" + destinationName + ", initKafkaProducer=" + initKafkaProducer + ", properties=" + properties);
            return initKafkaProducer;
        }
    }

    /**
     * 消费消息后提交
     */
    protected void commitOffsetSync(KafkaConsumer<String, Bytes> kafkaConsumer, Integer messageCount) {
        if (messageCount < 1) {
            log.debug("#112 not commitOffset, kafkaConsumer={}, messageCount={}", kafkaConsumer, messageCount);
            return;
        }

        Object autoCommit = this.getProperties().get(ENABLE_AUTO_COMMIT);
        if (null != autoCommit && "true".equals(autoCommit.toString())) {
            log.debug("#118 not commitOffset, kafkaConsumer={}, messageCount={}, autoCommit={}", kafkaConsumer, messageCount, autoCommit);
            return; // 已自动提交
        }

        kafkaConsumer.commitSync(); // 手动提交
        log.debug("#123 after commitOffset, kafkaConsumer={}, messageCount={}, autoCommit={}", kafkaConsumer, messageCount, autoCommit);
    }

    @Override
    public void destroy() throws Exception {
        for (KafkaProducer<String, Bytes> kafkaProducer : _KAFKA_PRODUCER_MAP.values()) {
            log.warn("#124 destroy, this=" + this + ", close kafkaProducer=" + kafkaProducer);
            kafkaProducer.close();
        }

        for (KafkaConsumer<String, Bytes> kafkaConsumer : _KAFKA_CONSUMER_MAP.values()) {
            log.warn("#124 destroy, this=" + this + ", close kafkaConsumer=" + kafkaConsumer);
            kafkaConsumer.close();
        }

        log.warn("#124 destroy, this=" + this + " done");
    }

    @SuppressWarnings({ "rawtypes", "unchecked" })
    public void setProperties(Map properties) {
        this.properties.putAll(properties); // 覆盖默认值
    }

    /**
     * 别改
     */
    public Map<String, Object> getProperties() {
        return Collections.unmodifiableMap(this.properties);
    }
}